package com.xdja.kafka.file.source;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceConnector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * kafka从文件读取数据到kafka的topic的connector类
 */
public class FileSourceConnector extends SourceConnector {
    // 定义主题配置变量
    public static final String TOPIC_CONFIG = "topic";    ///
    // 定义文件配置变量
    public static final String FILE_CONFIG = "file";

    // 实例化一个配置对象
    private static final ConfigDef CONFIG_DEF = new ConfigDef()
            .define(FILE_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "Source filename.")
            .define(TOPIC_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "The topic to publish data to");

    // 声明文件名变量
    private String filename;    //这是要读取的配置信息,从    curl命令的参数中读
    // 声明主题变量
    private String topic;

    /**
     * 构造方法
     */
    public FileSourceConnector(){
        System.out.println(  "FileSourceConnector构造方法");
    }

    /**
     * 获取当前版本
     * @return
     */
    @Override
    public String version() {
        String version= AppInfoParser.getVersion();
        System.out.println("调用了FileSourceConnector的 version() "+version);
        return version;
    }


    /**
     * connector加载时，执行的内容
     * @param props
     */
    @Override
    public void start(Map<String, String> props) {
        System.out.println("调用了 ZyFileStreamSourceConnector 的 start() "+props);
        filename = props.get(FILE_CONFIG);  //读取配置项  file,获取文件名  =  props.get("file");  ->  foo.txt
        topic = props.get(TOPIC_CONFIG);   //读取配置项  topic,获取主题名  =  props.get("topic");  -> connect-file-test
        if (topic == null || topic.trim().length() == 0)  //如果没有指定主题，则异常
            throw new ConnectException("FileSourceConnector configuration must include 'topic' setting");
        if (topic.contains(","))  //只允许一个connect同时从一个文件中读取数据
            throw new ConnectException("FileSourceConnector should only have a single topic when used as a source.");
    }

    /**
     * 指定当前connector对应的task处理类
     * @return
     */
    @Override
    public Class<? extends Task> taskClass() {
        System.out.println("调用了FileSourceConnector的taskClass() ");
        return FileSourceTask.class;
    }

    /**
     * 读取配置，写入系统上下文config中
     * @param maxTasks
     * @return
     */
    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        System.out.println("调用了FileSourceConnector的taskConfigs() "+maxTasks);

        ArrayList<Map<String, String>> configs = new ArrayList<Map<String, String>>();

        Map<String, String> config = new HashMap<String, String>();
        if (filename != null) {
            config.put(FILE_CONFIG, filename);
        }
        config.put(TOPIC_CONFIG, topic);
        configs.add(config);
        return configs;

    }

    /**
     * connector停止时，执行的内容
     */
    @Override
    public void stop() {
        System.out.println("关闭FileSourceConnector");
    }

    /**
     * 返回connector配置
     * @return
     */
    @Override
    public ConfigDef config() {
        System.out.println("调用FileSourceConnector的config()");
        return CONFIG_DEF;
    }
}
